
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.graphx.GraphXUtils.registerKryoClasses
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

import java.util
import java.util.Properties

object KafkaDStream {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.WARN)

    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName(this.getClass.getCanonicalName)
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");  //使用Kryo序列化库


    //conf.registerKryoClasses(Array(getClass(org.apache.kafka.clients.consumer.ConsumerRecord)))
    val ssc = new StreamingContext(conf, Seconds(2))

    // 定义kafka消费者相关参数
    val kafkaParams: Map[String, Object] = getKafkaConsumerParams()
    val topics: Array[String] = Array("spark_topic01")



    // 定义 kafka发送者相关参数
    val brokers = "linux128:9092,linux129:9092,linux130:9092"
    val topic2 = "spark_topic02"
    val prop = new Properties()

    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])

    // KafkaProducer
    val producer = new KafkaProducer[String, String](prop)

    // 从 kafka中获取数据
    // PreferBrokers: Use this only if your executors are on the same nodes as your Kafka brokers.
    // PreferConsistent: Use this in most cases, it will consistently distribute partitions across all executors.
    // PreferFixed: Use this to place particular TopicPartitions on particular hosts if your load is uneven.
    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))
    // Subscribe 或 SubscribePattern，订阅topic；有自动分区发现功能(运行流期间添加分区)
    // Assign 指定一个固定的分区集合。无自动分区发现功能【没有特殊需求少用】



    var i = 0;
    // DStream输出
    dstream.foreachRDD{(rdd, time) =>
      if (!rdd.isEmpty()) {
        println(s"*********** rdd.count = ${rdd.count()}; time = $time ***********")
        //rdd.foreach(str=>{
        rdd.collect().foreach (str=>{
          println("beforeHandledStr=>"+str.value().toString)
          var afterHandleStr = createObj(str.value().toString)
          println("afterHandleStr=>"+afterHandleStr)
          //再往另一个topic发送
          val msg1 = new ProducerRecord[String, String](topic2, i.toString,afterHandleStr)
          i = i+1
          producer.send(msg1)
          Thread.sleep(100)
        })

      }else{
        println("nothing to do")
      }
    }
    //producer.close()
    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaConsumerParams(groupId: String = "mygroup1"): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux128:9092,linux129:9092,linux130:9092",
      // org.apache.kafka.common.serialization.StringDeserializer
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean))
  }

  def createObj(line:String):String = {
    val fields :Array[String] = line.split(",")
    var subPhrase1 = fields(0).split("<<<!>>>")
    var commandid = ""
    if (subPhrase1.length>0){
      commandid = subPhrase1(1)
    }

    var subPhrase2 = fields(1).split("<<<!>>>")
    var houseid  = ""
    if (subPhrase2.length>0){
      houseid = subPhrase2(1)
    }


    var subPhrase3 = fields(2).split("<<<!>>>")
    var gathertime   = ""
    if (subPhrase3.length>0){
      gathertime  = subPhrase3(1)
    }

    var subPhrase4 = fields(3).split("<<<!>>>")
    var srcip    = ""
    if (subPhrase4.length>0){
      srcip   = subPhrase4(1)
    }

    var subPhrase5 = fields(4).split("<<<!>>>")
    var destip    = ""
    if (subPhrase5.length>0){
      destip   = subPhrase5(1)
    }

    var subPhrase6 = fields(5).split("<<<!>>>")
    var srcport    = ""
    if (subPhrase6.length>0){
      srcport   = subPhrase6(1)
    }

    var subPhrase7 = fields(6).split("<<<!>>>")
    var destport    = ""
    if (subPhrase7.length>0){
      destport   = subPhrase7(1)
    }

    var subPhrase8 = fields(7).split("<<<!>>>")
    var domainname    = ""
    if (subPhrase8.length>0){
      domainname   = subPhrase8(1)
    }

    var subPhrase9 = fields(8).split("<<<!>>>")
    var proxytype1     = ""
    if (subPhrase9.length>0){
      proxytype1   = subPhrase9(1)
    }

    var subPhrase10 = fields(9).split("<<<!>>>")
    var proxyip      = ""
    if (subPhrase10.length>0){
      proxyip    = subPhrase10(1)
    }

    var subPhrase11 = fields(10).split("<<<!>>>")
    var proxytype2     = ""
    if (subPhrase11.length>0){
      proxytype2   = subPhrase11(1)
    }

    var subPhrase12 = fields(11).split("<<<!>>>")
    var title      = ""
    if (subPhrase12.length>0){
      title    = subPhrase12(1)
    }

    var subPhrase13 = fields(12).split("<<<!>>>")
    var content       = ""
    if (subPhrase13.length>0){
      content     = subPhrase13(1)
    }

    var subPhrase14 = fields(13).split("<<<!>>>")
    var url       = ""
    if (subPhrase14.length>0){
      url     = subPhrase14(1)
    }

    var subPhrase15 = fields(14).split("<<<!>>>")
    var logid      = ""
    if (subPhrase15.length>0){
      logid    = subPhrase15(1)
    }
    commandid+"|"+houseid+"|"+gathertime+"|"+srcip+"|"+destip+"|"+srcport+"|"+destport+"|"+domainname+"|"+proxytype1+"|"+proxyip+"|"+proxytype2+"|"+title+"|"+content+"|"+url+"|"+logid
  }

}